package com.atguigu.bigdata.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bigdata.gmall.realtime.app.BaseAppV1;
import com.atguigu.bigdata.gmall.realtime.bean.TradeSkuOrderBean;
import com.atguigu.bigdata.gmall.realtime.common.Constant;
import com.atguigu.bigdata.gmall.realtime.util.AtguiguUtil;
import com.atguigu.bigdata.gmall.realtime.util.DimUtil;
import com.atguigu.bigdata.gmall.realtime.util.JdbcUtil;
import com.atguigu.bigdata.gmall.realtime.util.RedisUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.math.BigDecimal;
import java.sql.Connection;
import java.time.Duration;

/**
 * @Author lzc
 * @Date 2022/10/18 10:36
 */
public class Dws_09_DwsTradeSkuOrderWindow_Cache extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_09_DwsTradeSkuOrderWindow_Cache().init(
            4009,
            2,
            "Dws_09_DwsTradeSkuOrderWindow_Cache",
            Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
        
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          DataStreamSource<String> stream) {
        // 1. 按照 订单详情 id 去重
        SingleOutputStreamOperator<JSONObject> distinctedStream = distinct(stream);
        // 2. 把数据封装到 pojo 中
        SingleOutputStreamOperator<TradeSkuOrderBean> beanStream = parseToPOJO(distinctedStream);
        // 3. 开窗聚合
        SingleOutputStreamOperator<TradeSkuOrderBean> aggregatedStreamWithoutDims = windowAndAgg(beanStream);
        
        // 4. 补充维度信息
        addDims(aggregatedStreamWithoutDims);
        
        // 5. 写出到clickhouse 中
    }
    
    private void addDims(SingleOutputStreamOperator<TradeSkuOrderBean> stream) {
        // 补充维度信息:
        stream
            .map(new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
    
                private Jedis jedis;
                private Connection conn;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    conn = JdbcUtil.getPhoenixConnection();
                    jedis = RedisUtil.getRedisClient();
                }
    
                @Override
                public void close() throws Exception {
                    JdbcUtil.close(conn);
                    RedisUtil.close(jedis);
                }
    
                @Override
                public TradeSkuOrderBean map(TradeSkuOrderBean bean) throws Exception {
                    // 根据一个 id 去对应的维度表中查询
                    // 1. sku_info
                    JSONObject skuInfo = DimUtil.readDim(jedis, conn, "dim_sku_info", bean.getSkuId());
                    bean.setSkuName(skuInfo.getString("SKU_NAME"));
                    
                    bean.setSpuId(skuInfo.getString("SPU_ID"));
                    bean.setTrademarkId(skuInfo.getString("TM_ID"));
                    bean.setCategory3Id(skuInfo.getString("CATEGORY3_ID"));
                    
                    // 2. spu_info
                    JSONObject spuInfo = DimUtil.readDim(jedis, conn, "dim_spu_info", bean.getSpuId());
                    bean.setSpuName(spuInfo.getString("SPU_NAME"));
                    
                    // 3. base_trademark
                    JSONObject tm = DimUtil.readDim(jedis, conn, "dim_base_trademark", bean.getTrademarkId());
                    bean.setTrademarkName(tm.getString("TM_NAME"));
                    
                    
                    // 4. c3
                    JSONObject c3 = DimUtil.readDim(jedis, conn, "dim_base_category3", bean.getCategory3Id());
                    bean.setCategory3Name(c3.getString("NAME"));
                    bean.setCategory2Id(c3.getString("CATEGORY2_ID"));
                    
                    // 4. c2
                    JSONObject c2 = DimUtil.readDim(jedis, conn, "dim_base_category2", bean.getCategory2Id());
                    bean.setCategory2Name(c2.getString("NAME"));
                    bean.setCategory1Id(c2.getString("CATEGORY1_ID"));
                    
                    
                     // 4. c1
                    JSONObject c1 = DimUtil.readDim(jedis, conn, "dim_base_category1", bean.getCategory1Id());
                    bean.setCategory1Name(c1.getString("NAME"));
                    
                    return bean;
                }
            })
            .print();
    }
    
    private SingleOutputStreamOperator<TradeSkuOrderBean> windowAndAgg(
        SingleOutputStreamOperator<TradeSkuOrderBean> beanStream) {
        return beanStream
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<TradeSkuOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((bean, ts) -> bean.getTs())
            )
            .keyBy(TradeSkuOrderBean::getSkuId)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .reduce(
                new ReduceFunction<TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean b1,
                                                    TradeSkuOrderBean b2) throws Exception {
                        b1.setOrderOriginTotalAmount(b1.getOrderOriginTotalAmount().add(b2.getOrderOriginTotalAmount()));
                        b1.setOrderAmount(b1.getOrderAmount().add(b2.getOrderAmount()));
                        b1.setOrderActivityReduceAmount(b1.getOrderActivityReduceAmount().add(b2.getOrderActivityReduceAmount()));
                        b1.setOrderCouponReduceAmount(b1.getOrderCouponReduceAmount().add(b2.getOrderCouponReduceAmount()));
                        return b1;
                    }
                },
                new ProcessWindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void process(String skuId,
                                        Context ctx,
                                        Iterable<TradeSkuOrderBean> elements,
                                        Collector<TradeSkuOrderBean> out) throws Exception {
                        TradeSkuOrderBean bean = elements.iterator().next();
                        bean.setStt(AtguiguUtil.toDatTime(ctx.window().getStart()));
                        bean.setEdt(AtguiguUtil.toDatTime(ctx.window().getEnd()));
                        
                        //                        bean.setTs(System.currentTimeMillis());
                        bean.setTs(ctx.currentProcessingTime());
                        
                        out.collect(bean);
                        
                    }
                }
            
            );
    }
    
    private SingleOutputStreamOperator<TradeSkuOrderBean> parseToPOJO(
        SingleOutputStreamOperator<JSONObject> distinctedStream) {
        return distinctedStream
            .map(obj -> TradeSkuOrderBean.builder()
                .skuId(obj.getString("sku_id"))
                .orderOriginTotalAmount(obj.getBigDecimal("split_original_amount"))
                .orderAmount(obj.getBigDecimal("split_total_amount"))
                .orderActivityReduceAmount(obj.getBigDecimal("split_activity_amount") == null ? new BigDecimal("0") : obj.getBigDecimal("split_activity_amount"))
                .orderCouponReduceAmount(obj.getBigDecimal("split_coupon_amount") == null ? new BigDecimal("0") : obj.getBigDecimal("split_coupon_amount"))
                .ts(obj.getLong("ts") * 1000) // 把s 变成 ms
                .build());
    }
    
    private SingleOutputStreamOperator<JSONObject> distinct(DataStreamSource<String> stream) {
        /*
        去重思路:
            目的 要 row_op_ts 最大的那个, 最全的, 也一定是最后来的那个
            
        思路 1:
            没有办法从个数上记录, 找到最后一个
            
            数据是同时产生, 由于网络的波动导致的数据重复
            不管详情有多少条重复, 最终 5s 内一定会来齐
            
            第一个来的的时候, 注册一个定时器:5s 后触发的定时器(处理时间)
                当定时器触发的时候, 同一个详情的所有重复数据来齐.
                找到row_op_ts最大的那个
                提高效率: 每来一个都比较得到最大的那个
                
            
           最终结果: 第一个来了之后, 5s 后处结果
                
        思路 2:
            窗口: 把同一个详情的重复数据, 放入到同一个窗口中, 等到窗口触发计算的时候, 排序找到最大的.
            
            基于时间的窗口: 滚动的基于事件时间, 窗口长度: 1s
                会话窗口: gap 5s
                    最后来了之后,等待 5s 之后出结果
                    
        思路 3:
            如果要汇总的指标只在左表中, 有表的没用.无需等最完整的那条数据
            
            换句话说过, 第一条不管全不全, 都满足需求. 所以只取第一条
                
            
            
         */
        return stream
            .map(JSON::parseObject)
            .keyBy(obj -> obj.getString("id"))
            // 要用定时器: 必须先 keyBy
            .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                
                private ValueState<JSONObject> dataState;
                private ValueState<Boolean> isFirstState;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    isFirstState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("isFirstState", Boolean.class));
                    dataState = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("dataState", JSONObject.class));
                }
                
                @Override
                public void onTimer(long timestamp,
                                    KeyedProcessFunction<String, JSONObject, JSONObject>.OnTimerContext ctx,
                                    Collector<JSONObject> out) throws Exception {
                    // 定时器触发的时候执行这个方法.
                    // 当定时器触发的时候, 状态中就是最新的那个
                    // 把状态中的值输出
                    out.collect(dataState.value());
                }
                
                @Override
                public void processElement(JSONObject value,
                                           Context ctx,
                                           Collector<JSONObject> out) throws Exception {
                    
                    // 当这个详情的第一条数据进来的时候, 注册一个定时器: 5s 后触发的定时器
                    if (isFirstState.value() == null) {
                        // 更新状态
                        isFirstState.update(false);
                        ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime() + 5000);
                        // 存储数据:
                        dataState.update(value);
                    } else {
                        // 不是第一条: 当前和状态中的进行比较, 时间大的存入到状态中
                        /*
                        2022-10-11 07:42:48.491Z
                        2022-10-11 07:42:48.49Z  // 490
                        2022-10-11 07:42:48.5Z   // 500ms
                        2022-10-11 07:42:48.005Z   // 5ms
                        
                        2022-10-11 07:42:48.51
                        2022-10-11 07:42:48.5
                         */
                        String last = dataState.value().getString("row_op_ts").replaceAll("Z", "");
                        String current = value.getString("row_op_ts").replaceAll("Z", "");
                        // 新数据的时间大于旧数据, 就更新状态
                        if (current.compareTo(last) > 0) {
                            dataState.update(value);
                        }
                    }
                }
            });
    }
}
/*
对热点数据, 应该缓存起来, 避免频繁的查询数据库

缓存指的内存

1. 堆内存
    flink 的状态
    
    好处: 本地内存, 存取速度极快, 不需要通过网络. 数据结构丰富, 可以选择需要的数据结构
    坏处: 1. 占用 flink 的本地内存, 影响计算.
         2. 当维度发生变化的时候, 状态中缓存的维度不能及时更新
    
2. 外置内存: redis
    坏处: 外置内存, 访问都需要通过网络.
    
    好处: 维度发生变化的时候, 缓存的维度能够及时及时更新

最终选择 redis 旁路缓存

5大数据结构, 选择什么?
select * from t where id='1'
string
    key                     value
    表名+id                  json 格式的字符串
    dim_sku_info:1          {"id": "", ....}
    
    好处:
        1. 存取及其方便
        2. 可以单独的给每条维度信息设置 ttl
        
    坏处:
        占用大量的 key, 不方便管理. key 越大越容易冲突
        0-15 库, 单独选一个没有人使用的库

list
    key                     value
    表名                    [{}, {}]
    dim_sku_info            ....
    
    好处: key 少, 一张表一个 key
    坏处:
        1. 查询需要遍历, 效率比较低
        2. 如果给 key 设置的 ttl, 这个表的所有缓存的维度会一起失效

set
  相比 list 多了一个去重的功能

hash(map)
    key                 field       value
    dim_sku_info        1           {json 格式的字符串}
                        ....
                        
    好处:
        1. 读写方便  hset(key, field, value)
        2. key 比较少, 一个表一个 key
        
    坏处:
        一个张表的维度数据, 共用一个 key., ttl 也是只能设置到 key.
        没有办法单独的给每条维度数据设置 ttl

zset
    不用
    

经过对比: string 比较合适

key:   表名:id
value: json 格式的字符串
ttl:   2 天



-----------
交易域SKU粒度下单各窗口汇总表

统计各维度各窗口的原始金额、活动减免金额、优惠券减免金额和订单金额

1. 数据源
    dwd 下单事务事实表
        粒度: sku
        
        order_detail
            内连接
        order_info
            左连接
        详情活动
            左连接
        详情优惠券
            lookup 的内连接
        字典表
2. 数据源有左连接, 需要去重
     
     详情1 sku1  100    null
     null
     详情1 sku1  100     20
     
     去重: 按照详情 id 去重, 保留最全的那个. 时间最大的那个
 
3. 把每条数据封装到 pojo 中

4. 按照 sku 分组开窗聚合
   0-5 sku1  100 200 ...

5. 聚合后, 根据 sku_id 去补充其他的维度信息
    
    根据 sku_id 去查找对应的维度: sku_info
    
    select * from sku_info where id=1
        使用 jdbc 去查找
        
        优化: 旁路缓存和异步
        
6. 写出到 clickhouse 中
 
 
 
 #!/bin/bash
function my_sleep(){
    for(( sec=$1;sec>=0;sec-- ))
    do
        echo -ne "\e[1;31m $sec $2\e[0m"
        echo -ne "\r"
        sleep 1
    done
    echo ''
}
zk stop
zk start
hadoop.sh start
redis-server /etc/redis.conf
maxwell.sh start
my_sleep 20 "秒后继续启动其他集群..."
flume.sh start
kafka.sh start
start-hbase.sh
/opt/module/flink-1.13.6/bin/yarn-session.sh -d

 */